package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/5/1 21:02
 */
public class DWD_13_DwdToolCouponPayApp extends BaseSQLApp {
    public static void main(String[] args) {
        new DWD_13_DwdToolCouponPayApp().init("DwdToolCouponPayApp", 3013, 1, "DwdToolCouponPayApp");
        
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tEnv) {
        /*
            更新
            并且 used_time 不是null
            并且 old['coupon_status'] ='1402'
            
        
         */
        // TODO 3. 从 Kafka 读取业务数据，封装为 Flink SQL 表
        tEnv.executeSql(
            "create table `topic_db` ( " +
                "`database` string, " +
                "`table` string, " +
                "`data` map<string, string>, " +
                "`type` string, " +
                "`old` string, " +
                "`ts` string " +
                ")" + SQLUtil.getKafkaDDL("dwd_tool_coupon_pay", "ods_db"));
        
        // TODO 4. 读取优惠券领用表数据，筛选优惠券使用（支付）数据
        Table couponUsePay = tEnv.sqlQuery(
            "select " +
                "data['id'] id, " +
                "data['coupon_id'] coupon_id, " +
                "data['user_id'] user_id, " +
                "data['order_id'] order_id, " +
                "date_format(data['used_time'],'yyyy-MM-dd') date_id, " +
                "data['used_time'] used_time, " +
                "`old`, " +
                "ts " +
                "from topic_db " +
                "where `table` = 'coupon_use' " +
                "and `type` = 'update' " +
                "and data['used_time'] is not null " +
                "and old['coupon_stats']='1402'");
        
        tEnv.createTemporaryView("coupon_use_pay", couponUsePay);
        
        // TODO 5. 建立 Upsert-Kafka dwd_tool_coupon_order 表
        tEnv.executeSql(
            "create table dwd_tool_coupon_pay( " +
                "id string, " +
                "coupon_id string, " +
                "user_id string, " +
                "order_id string, " +
                "date_id string, " +
                "payment_time string, " +
                "ts string, " +
                "primary key(id) not enforced " +
                ")" + SQLUtil.getUpsertKafkaDDL("dwd_tool_coupon_pay"));
        
        // TODO 6. 将数据写入 Upsert-Kafka 表
        tEnv.executeSql(
            "insert into dwd_tool_coupon_pay select " +
                "id, " +
                "coupon_id, " +
                "user_id, " +
                "order_id, " +
                "date_id, " +
                "used_time payment_time, " +
                "ts from coupon_use_pay");
    }
}
